package com.ruyuan.engine.main;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.RuleResult;
import com.ruyuan.engine.pojo.TriggerTimer;
import com.ruyuan.engine.pojo.advance.*;
import com.ruyuan.engine.router.RouterService;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.kie.api.runtime.KieSession;

import java.util.List;

public class RuleProcessFunction extends KeyedBroadcastProcessFunction<String, MultipleConditionPartitionRule, RuleBroadcastTableEnhance, RuleResult> {

    public static MapStateDescriptor<String, List<RuleComplexBeanEnhance> > ruleStateDesc = new MapStateDescriptor<>("rule_broadcast_state",TypeInformation.of(String.class),TypeInformation.of(new TypeHint<List<RuleComplexBeanEnhance>>() {}));
    ListState<ClientLog> clientLogListState;
    //定时信息规则State
    ListState<Tuple2<RuleComplexBeanEnhance, Long>> timerState;
    ReadOnlyBroadcastState<String, List<RuleComplexBeanEnhance>> broadcastState;
    //核心查询服务
    RouterService routerService;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化State
        ListStateDescriptor<ClientLog> listStateDescriptor = new ListStateDescriptor<ClientLog>("clientLogListState",ClientLog.class);
        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.hours(6)).updateTtlOnCreateAndWrite().build();
        listStateDescriptor.enableTimeToLive(stateTtlConfig);
        clientLogListState = getRuntimeContext().getListState(listStateDescriptor);
        routerService = new RouterService(clientLogListState);

        //是否定时
        ListStateDescriptor<Tuple2<RuleComplexBeanEnhance, Long>> ruleTimerStateDesc =  new ListStateDescriptor<Tuple2<RuleComplexBeanEnhance, Long>>("rule_timer", TypeInformation.of(new TypeHint<Tuple2<RuleComplexBeanEnhance, Long>>() {}));
        timerState = getRuntimeContext().getListState(ruleTimerStateDesc);
    }

    /**
     * 针对数据的处理规则
     * */
    @Override
    public void processElement(MultipleConditionPartitionRule multipleConditionPartitionRule, ReadOnlyContext ctx, Collector<RuleResult> out) throws Exception {

        broadcastState = ctx.getBroadcastState(ruleStateDesc);
        //获取对应的规则
        List<RuleComplexBeanEnhance> ruleComplexBeanEnhances = broadcastState.get(multipleConditionPartitionRule.getPartitionKeys());
        //遍历状态对应的规则
        for(RuleComplexBeanEnhance ruleComplexBeanEnhance:ruleComplexBeanEnhances) {
            RuleComplexBean ruleComplexBean = ruleComplexBeanEnhance.getRuleComplexBean();
            KieSession kieSession = ruleComplexBeanEnhance.getKieSession();
            /**
             * 组织规则，组织数据，组织查询服务进行绑定，具体逻辑在drools控制。
             * */
            RuleFactBase ruleFactBase = new RuleFactBase(ruleComplexBean,routerService,multipleConditionPartitionRule.getClientLog(),false);
            //调用drools引擎进行计算
            kieSession.insert(ruleFactBase);
            kieSession.fireAllRules();
            //获取规则计算的结果
            boolean match = ruleFactBase.isMatch();
            //如果结果为true，判断是否有定时条件
            if (match) {
                if (ruleComplexBean.isRegisterTimer()) {
                    TriggerTimer triggerTimer = ruleComplexBean.getTriggerTimers().get(0);
                    long timeTrigger = multipleConditionPartitionRule.getClientLog().getCreateTime() + triggerTimer.getTimeLimit();
                    ctx.timerService().registerEventTimeTimer(timeTrigger);
                    timerState.add(Tuple2.of(ruleComplexBeanEnhance,timeTrigger));
                }else {
                    //没有定时条件则输出最终结果
                    RuleResult ruleResult = new RuleResult();
                    ruleResult.setRuleId(ruleComplexBean.getRuleId());
                    ruleResult.setRulePartitionValue(multipleConditionPartitionRule.getPartitionValues());
                    ruleResult.setCreateTime(DateFormatUtils.format(System.currentTimeMillis(),"yyyy-MM-dd hh:mm:ss"));
                    ruleResult.setIsMatch(match);
                    ruleResult.setUserId(multipleConditionPartitionRule.getClientLog().getUserNo());
                    out.collect(ruleResult);
                }
            }
        }
    }

    @Override
    public void processBroadcastElement(RuleBroadcastTableEnhance value, Context ctx, Collector<RuleResult> out) {

    }
}